/**
 * Copyright © 2017 The Thingsboard Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.thingsboard.server.service.remote;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Function;
import com.google.common.util.concurrent.FutureCallback;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.thingsboard.rule.engine.api.RpcError;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
import org.thingsboard.server.common.data.kv.*;
import org.thingsboard.server.common.data.rpc.RpcRequest;
import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody;
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.remote.upload.RpcCommandListener;
import org.thingsboard.server.remote.upload.conf.TbGatewayConfiguration;
import org.thingsboard.server.remote.upload.data.*;
import org.thingsboard.server.remote.upload.gateway.GatewayService;
import org.thingsboard.server.remote.upload.gateway.MqttDeliveryFuture;
import org.thingsboard.server.remote.upload.gateway.MqttGatewaySecurityConfiguration;
import org.thingsboard.server.remote.upload.util.JsonTools;
import org.thingsboard.server.service.rpc.TbCoreDeviceRpcService;
import org.thingsboard.server.service.security.AccessValidator;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;

import javax.annotation.Nullable;
import javax.annotation.PostConstruct;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static org.thingsboard.server.remote.upload.util.JsonTools.*;


/**
 *  核心网关实现
 *  GatewayService  tb-core消息传递调用
 *  MqttCallback    处理断开重连
 *  IMqttMessageListener  接收消息
 *
 */
@Service
@Slf4j
public class DefaultMqttGatewayService implements GatewayService, MqttCallback, IMqttMessageListener {

    /**
     * 客户端连接重新检查周期
     */
    private static final long CLIENT_RECONNECT_CHECK_INTERVAL = 1;

    /**
     * 订阅
     *  设备（网关本身） 共享属性变化
     */
    public static final String DEVICE_GET_ATTRIBUTES_REQUEST_TOPIC = "v1/devices/me/attributes/request/1";
    /**
     *  发布
     *  设备（网关本身） 共享属性变化
     */
    public static final String DEVICE_GET_ATTRIBUTES_RESPONSE_TOPIC = "v1/devices/me/attributes/response/1";
    /**
     *订阅
     *  设备（网关本身） 共享属性请求
     */
    public static final String DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC = "v1/devices/me/attributes/response/+";

    /**
     *   设备名称——设备信息（名称 类型）
     */
    private final ConcurrentMap<String, DeviceInfo> devices = new ConcurrentHashMap<>();

    /**
     * 属性更新上传的数量
     *  一个属性上传一次+1
     */
    private final AtomicLong attributesCount = new AtomicLong();
    /**
     * 遥测更新上传 计数
     */
    private final AtomicLong telemetryCount = new AtomicLong();
    /**
     * 请求ID 自增生成器
     */
    private final AtomicInteger msgIdSeq = new AtomicInteger();
    /**
     * 属性更新订阅
     */
    private final Set<AttributesUpdateSubscription> attributeUpdateSubs = ConcurrentHashMap.newKeySet();
    /**
     *Rpc 订阅
     */
    private final Set<RpcCommandSubscription> rpcCommandSubs = ConcurrentHashMap.newKeySet();
    /**
     * 请求
     */
    private final Map<AttributeRequestKey, AttributeRequestListener> pendingAttrRequestsMap = new ConcurrentHashMap<>();

    @Autowired
    private TbGatewayConfiguration configuration;
    @Autowired
    protected TelemetrySubscriptionService tsSubService;

    @Autowired
    protected DeviceService deviceService;
    private volatile ObjectNode error;
    @Value("${server.rest.server_side_rpc.min_timeout:5000}")
    private long minTimeout;

    @Value("${server.rest.server_side_rpc.default_timeout:10000}")
    private long defaultTimeout;
    /**
     * eclipse.paho
     */
    private MqttAsyncClient tbClient;
    /**
     * eclipse.paho
     */
    private MqttConnectOptions tbClientOptions;

    private Object connectLock = new Object();


    protected final ObjectMapper jsonMapper = new ObjectMapper();
    /**
     * 定时任务执行器
     */
    private ScheduledExecutorService scheduler;
    private ExecutorService callbackExecutor = Executors.newCachedThreadPool();
    @Autowired
    private TbCoreDeviceRpcService deviceRpcService;
    /**
     * 初始化方法
     *  MQTT连接配置
     *   准备
     *   注册回调方法
     * @throws Exception
     */
    @PostConstruct
    @Override
    public void init() throws Exception {
        if(configuration==null){
            log.warn("远程平台连接失败.....");
            return;
        }
        if(!configuration.getConnect()){
            log.info("远程平台连接未开启");
            return;
        }
        scheduler = Executors.newSingleThreadScheduledExecutor();
        tbClientOptions = new MqttConnectOptions();
        tbClientOptions.setCleanSession(false);
        tbClientOptions.setMaxInflight(configuration.getConnection().getMaxInFlight());
        tbClientOptions.setAutomaticReconnect(true);

        MqttGatewaySecurityConfiguration security = configuration.getConnection().getSecurity();
        security.setupSecurityOptions(tbClientOptions);

        //eclipse.paho 异步MQTT客户端
        // 回调方法接口 MqttCallback
        tbClient = new MqttAsyncClient((security.isSsl() ? "ssl" : "tcp") + "://" + configuration.getConnection().getHost() + ":" + configuration.getConnection().getPort(),
                security.getClientId(), configuration.getPersistence().getPersistence());
        tbClient.setCallback(this);

        // 持久化策略
        if (configuration.getPersistence().getBufferSize() > 0) {
            DisconnectedBufferOptions options = new DisconnectedBufferOptions();
            options.setBufferSize(configuration.getPersistence().getBufferSize());
            options.setBufferEnabled(true);
            options.setPersistBuffer(true);
            tbClient.setBufferOpts(options);
        }
        connect();

          // 定时 发布 网关本身一些遥测数据 处理指标
        scheduler.scheduleAtFixedRate(this::reportStats, 0, configuration.getReporting().getInterval(), TimeUnit.MILLISECONDS);
    }

    /**
     * 销毁方法
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        scheduler.shutdownNow();
        callbackExecutor.shutdownNow();
        tbClient.disconnect();
    }

    @Override
    public String getTenantLabel() {
        return configuration.getLabel();
    }

    @Override
    public boolean isUsable() {
        return configuration != null && configuration.getConnect();
    }


    /**
     * 外部调用方法
     * 处理和发布
     *   设备接入网关
     * @param deviceName
     * @param deviceType
     * @return
     */
    @Override
    public MqttDeliveryFuture onDeviceConnect(String deviceName, String deviceType) {
        final int msgId = msgIdSeq.incrementAndGet();
        deviceName=fullName(deviceName);
        ObjectNode node = newNode().put("device", deviceName);
        if (!StringUtils.isEmpty(deviceType)) {
            node.put("type", deviceType);
        }
        MqttMessage msg = new MqttMessage(toBytes(node));
        msg.setId(msgId);
//        log.info("[{}] - [{}]Device Connected!", deviceName,deviceType);
        devices.putIfAbsent(deviceName, new DeviceInfo(deviceName, deviceType));
        String finalDeviceName = deviceName;
        return publishAsync(MqttTopics.GATEWAY_CONNECT_TOPIC, msg,
                token -> {
                    log.info("[{}][{}] Device connect event is reported to Thingsboard!", finalDeviceName, msgId);
                },
                error -> log.warn("[{}][{}] Failed to report device connection!", finalDeviceName, msgId, error));
    }


    /**
     * 外部调用方法
     *  处理发布
     *   网关 有设备掉线的请求
     * @param deviceName
     * @return
     */
    @Override
    public Optional<MqttDeliveryFuture> onDeviceDisconnect(String deviceName) {
        deviceName=fullName(deviceName);
        if (devices.remove(deviceName) != null) {
            final int msgId = msgIdSeq.incrementAndGet();
            byte[] msgData = toBytes(newNode().put("device", deviceName));
            MqttMessage msg = new MqttMessage(msgData);
            msg.setId(msgId);
            log.info("[{}][{}] Device Disconnected!", deviceName, msgId);
            String finalDeviceName = deviceName;
            MqttDeliveryFuture future = publishAsync(MqttTopics.GATEWAY_DISCONNECT_TOPIC, msg,
                    token -> {
                        log.info("[{}][{}] Device disconnect event is delivered!", finalDeviceName, msgId);
                    },
                    error -> log.warn("[{}][{}] Failed to report device disconnect!", finalDeviceName, msgId, error));
            return Optional.of(future);
        } else {
            log.debug("[{}] Device was disconnected before. Nothing is going to happened.", deviceName);
            return Optional.empty();
        }
    }

    /**
     * 外部调用方法
     *   设备属性更新
     *   发布
     * @param deviceName - the device name
     * @param attributes - the attribute values list
     * @return
     */
    @Override
    public MqttDeliveryFuture onDeviceAttributesUpdate(String deviceName, List<KvEntry> attributes) {
        final int msgId = msgIdSeq.incrementAndGet();
        log.trace("[{}][{}] Updating device attributes: {}", deviceName, msgId, attributes);
        deviceName=fullName(deviceName);
        checkDeviceConnected(deviceName);

        ObjectNode node = newNode();
        ObjectNode deviceNode = node.putObject(deviceName);
        attributes.forEach(kv -> putToNode(deviceNode, kv));
        final int packSize = attributes.size();
        MqttMessage msg = new MqttMessage(toBytes(node));
        msg.setId(msgId);
        String finalDeviceName = deviceName;
        return publishAsync(MqttTopics.GATEWAY_ATTRIBUTES_TOPIC, msg,
                token -> {
                    log.debug("[{}][{}] Device attributes were delivered!", finalDeviceName, msgId);
                    attributesCount.addAndGet(packSize);
                },
                error -> log.warn("[{}][{}] Failed to report device attributes!", finalDeviceName, msgId, error));
    }

    /**
     * 外部调用方法
     *  设备遥测更新
     *  发布
     * @param deviceName - the device name
     * @param telemetry  - the telemetry values list
     * @return
     */
    @Override
    public MqttDeliveryFuture onDeviceTelemetry(String deviceName, List<TsKvEntry> telemetry) {
        final int msgId = msgIdSeq.incrementAndGet();
        log.trace("[{}][{}] Updating device telemetry: {}", deviceName, msgId, telemetry);
        deviceName=fullName(deviceName);
        checkDeviceConnected(deviceName);
        ObjectNode node = newNode();
        Map<Long, List<TsKvEntry>> tsMap = telemetry.stream().collect(Collectors.groupingBy(v -> v.getTs()));
        ArrayNode deviceNode = node.putArray(deviceName);
        tsMap.entrySet().forEach(kv -> {
            Long ts = kv.getKey();
            ObjectNode tsNode = deviceNode.addObject();
            tsNode.put("ts", ts);
            ObjectNode valuesNode = tsNode.putObject("values");
            kv.getValue().forEach(v -> putToNode(valuesNode, v));
        });
        final int packSize = telemetry.size();
        MqttMessage msg = new MqttMessage(toBytes(node));
        msg.setId(msgId);
        String finalDeviceName = deviceName;
        return publishAsync(MqttTopics.GATEWAY_TELEMETRY_TOPIC, msg,
                token -> {
                    log.debug("[{}][{}] Device telemetry published to ThingsBoard!", msgId, finalDeviceName);
                    telemetryCount.addAndGet(packSize);
                },
                error -> log.warn("[{}][{}] Failed to publish device telemetry!", finalDeviceName, msgId, error));
    }

    /**
     *  外部调用
     *   设备请求服务端共享属性
     *   发布
     * @param request
     * @param listener         - attributes response
     */
    @Override
    public void onDeviceAttributeRequest(AttributeRequest request, Consumer<AttributeResponse> listener) {
        final int msgId = msgIdSeq.incrementAndGet();
        String deviceName = request.getDeviceName();
        AttributeRequestKey requestKey = new AttributeRequestKey(request.getRequestId(), request.getDeviceName());
        log.trace("[{}][{}] Requesting {} attribute: {}", deviceName, msgId, request.isClientScope() ? "client" : "shared", request.getAttributeKey());
        checkDeviceConnected(deviceName);
        ObjectNode node = newNode();
        node.put("id", request.getRequestId());
        node.put("client", request.isClientScope());
        node.put("device", request.getDeviceName());
        node.put("key", request.getAttributeKey());
        MqttMessage msg = new MqttMessage(toBytes(node));

        msg.setId(msgId);
        pendingAttrRequestsMap.put(requestKey, new AttributeRequestListener(request, listener));
        publishAsync(MqttTopics.GATEWAY_ATTRIBUTES_REQUEST_TOPIC, msg,
                token -> {
                    log.debug("[{}][{}] Device attributes request was delivered!", deviceName, msgId);
                },
                error -> {
                    log.warn("[{}][{}] Failed to report device attributes!", deviceName, msgId, error);
                    pendingAttrRequestsMap.remove(requestKey);
                });
    }

    /**
     * 外部调用
     *  设备RPC处理返回
     *  发布
     * @param response - the device response to RPC call
     */
    @Override
    public void onDeviceRpcResponse(RpcCommandResponse response) {
        final int msgId = msgIdSeq.incrementAndGet();
        int requestId = response.getRequestId();
        String deviceName = response.getDeviceName();
        String data = response.getData();
        checkDeviceConnected(deviceName);

        ObjectNode node = newNode();
        node.put("id", requestId);
        node.put("device", deviceName);
        node.put("data", data);
        MqttMessage msg = new MqttMessage(toBytes(node));
        msg.setId(msgId);
        publishAsync(MqttTopics.GATEWAY_RPC_TOPIC, msg,
                token -> {
                    log.debug("[{}][{}] RPC response from device was delivered!", deviceName, requestId);
                },
                error -> {
                    log.warn("[{}][{}] Failed to report RPC response from device!", deviceName, requestId, error);
                });
    }

    @Override
    public boolean subscribe(AttributesUpdateSubscription subscription) {
        return subscribe(attributeUpdateSubs::add, subscription);
    }

    @Override
    public boolean subscribe(RpcCommandSubscription subscription) {
        return subscribe(rpcCommandSubs::add, subscription);
    }

    @Override
    public boolean unsubscribe(AttributesUpdateSubscription subscription) {
        return unsubscribe(attributeUpdateSubs::remove, subscription);
    }

    @Override
    public boolean unsubscribe(RpcCommandSubscription subscription) {
        return unsubscribe(rpcCommandSubs::remove, subscription);
    }

    @Override
    public void onError(Exception e) {
        onError(null, e);
    }

    @Override
    public void onError(String deviceName, Exception e) {
        ObjectNode node = newNode();
        node.put("ts", System.currentTimeMillis());
        if (deviceName != null) {
            node.put("device", deviceName);
        }
        node.put("error", toString(e));
        error = node;
    }

    /**
     *  连接管理
     *   真正连接
     *   断开重连
     *   订阅  DEVICE_ATTRIBUTES_TOPIC | DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC
     */
    private void connect() {
        if (!tbClient.isConnected()) {
            synchronized (connectLock) {
                while (!tbClient.isConnected()) {
                    log.debug("Attempt to connect to Thingsboard!");
                    try {
                        tbClient.connect(tbClientOptions, null, new IMqttActionListener() {
                            @Override
                            public void onSuccess(IMqttToken iMqttToken) {
                                log.info("Connected to Thingsboard!");
                            }

                            @Override
                            public void onFailure(IMqttToken iMqttToken, Throwable e) {
                            }
                        }).waitForCompletion();

                        tbClient.subscribe(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, 1, (IMqttMessageListener) this).waitForCompletion();

                        tbClient.subscribe(DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC, 1, (IMqttMessageListener) this).waitForCompletion();
                        ObjectNode node = newNode().put("shared", "configuration");
                        MqttMessage msg = new MqttMessage(toBytes(node));
                        tbClient.publish(DEVICE_GET_ATTRIBUTES_REQUEST_TOPIC, msg).waitForCompletion();
                        devices.forEach((k, v) -> onDeviceConnect(v.getName(), v.getType()));
                    } catch (Exception e) {
                        log.warn("Failed to connect to ThingsBoard!", e);
                        if (!tbClient.isConnected()) {
                            try {
                                Thread.sleep(configuration.getConnection().getRetryInterval());
                            } catch (InterruptedException e1) {
                                log.trace("Failed to wait for retry interval!", e);
                            }
                        }
                    }
                }
            }
        }
    }

    private void checkDeviceConnected(String deviceName) {
        if (!devices.containsKey(deviceName)) {
            onDeviceConnect(deviceName, null);
        }
    }

    /**
     *  异步发布
     * @param topic
     * @param msg
     * @param onSuccess
     * @param onFailure
     * @return
     */
    private MqttDeliveryFuture publishAsync(final String topic, MqttMessage msg, Consumer<IMqttToken> onSuccess, Consumer<Throwable> onFailure) {
        try {
            IMqttDeliveryToken token = tbClient.publish(topic, msg, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    onSuccess.accept(asyncActionToken);
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable e) {
                    onFailure.accept(e);
                }
            });
            return new MqttDeliveryFuture(token);
        } catch (MqttException e) {
            onFailure.accept(e);
            return new MqttDeliveryFuture(e);
        }
    }

    /**
     * 网关向 tb定时报告处理状态
     *   在线设备数
     *    已经上传的 ：属性计数 遥测计数
     *    发布 主题 DEVICE_TELEMETRY_TOPIC
     */
    private void reportStats() {
        if (tbClient == null) {
            log.info("Can't report stats because client was not initialized yet!");
            return;
        }
        ObjectNode node = newNode();
        node.put("ts", System.currentTimeMillis());
        ObjectNode valuesNode = node.putObject("values");

        valuesNode.put("devicesOnline", devices.size());
        valuesNode.put("attributesUploaded", attributesCount.getAndSet(0));
        valuesNode.put("telemetryUploaded", telemetryCount.getAndSet(0));
        if (error != null) {
            valuesNode.put("latestError", JsonTools.toString(error));
            error = null;
        }
        MqttMessage msg = new MqttMessage(toBytes(node));
        msg.setId(msgIdSeq.incrementAndGet());
        publishAsync(MqttTopics.DEVICE_TELEMETRY_TOPIC, msg,
                token -> log.info("Gateway statistics {} reported!", node),
                error -> log.warn("Failed to report gateway statistics!", error));
    }

    @Override
    public void connectionLost(Throwable cause) {
        //TODO: reply with error
        pendingAttrRequestsMap.clear();
        scheduler.schedule(this::checkClientReconnected, CLIENT_RECONNECT_CHECK_INTERVAL, TimeUnit.SECONDS);
    }

    /**
     *  订阅的消息处理
     * @param topic
     * @param message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.trace("Message arrived [{}] {}", topic, message.getId());
        log.warn("主题【{}】----【{}】",topic,fromString(new String(message.getPayload(), StandardCharsets.UTF_8)));
        callbackExecutor.submit(() -> {
            try {
                if (topic.equals(MqttTopics.GATEWAY_ATTRIBUTES_TOPIC)) {
                    onAttributesUpdate(message);
                } else if (topic.equals(MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC)) {
                    onDeviceAttributesResponse(message);
                } else if (topic.equals(MqttTopics.GATEWAY_RPC_TOPIC)) {
                    onRpcCommand(message);
                } else if (topic.equals(MqttTopics.DEVICE_ATTRIBUTES_TOPIC)) {
                    onGatewayAttributesUpdate(message);
                } else if (topic.equals(DEVICE_GET_ATTRIBUTES_RESPONSE_TOPIC)) {
                    onGatewayAttributesGet(message);
                }
            } catch (Exception e) {
                log.warn("Failed to process arrived message!", message);
            }
        });
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.trace("Delivery complete [{}]", token);
    }

    /**
     * 远程调用 属性更新执行
     * @param message
     */
    private void onAttributesUpdate(MqttMessage message) {

        JsonNode payload = fromString(new String(message.getPayload(), StandardCharsets.UTF_8));
        String deviceName = payload.get("device").asText();
        if(!devices.containsKey(deviceName)){
               return;
        }
        deviceName=realName(deviceName);
        Device device=deviceService.findDeviceByName(deviceName);
        if(device!=null){
            Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(payload.get("data").toString()));
            tsSubService.saveAndNotify(device.getTenantId(), device.getId(), "SHARED_SCOPE",new ArrayList<>(attributes) , new FutureCallback<Void>() {
                @Override
                public void onSuccess(@Nullable Void tmp) {
                }
                @Override
                public void onFailure(Throwable t) {
                     log.warn(t.toString());
                }
            });
        }
    }



    /**
     * RPC 执行
     *
     *
     * @param message
     */
    private void onRpcCommand(MqttMessage message) throws JsonProcessingException {
        RpcCommandResponse response=new RpcCommandResponse();
        JsonNode payload = fromString(new String(message.getPayload(), StandardCharsets.UTF_8));
        String deviceName = payload.get("device").asText();
        JsonNode data= payload.get("data");
        RpcRequest cmd = new RpcRequest(data.get("method").asText(),
                jsonMapper.writeValueAsString(data.get("params")));
        int id =  data.get("id").asInt();
        response.setDeviceName(deviceName);
        response.setRequestId(id);
        if(!devices.containsKey(deviceName)){
           response.setData("设备不存在");
           onDeviceRpcResponse( response);
          return;
        }
        if(StringUtils.isEmpty(cmd.getMethodName())||StringUtils.isEmpty(cmd.getRequestData())){
            response.setData("无效指令");
            onDeviceRpcResponse( response);
            return;
        }
        deviceName=realName(deviceName);
        Device device=deviceService.findDeviceByName(deviceName);
        if(device!=null){
            long expTime = System.currentTimeMillis() + Math.max(minTimeout, defaultTimeout);
            ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(cmd.getMethodName(), cmd.getRequestData());
            ToDeviceRpcRequest rpcRequest = new ToDeviceRpcRequest(UUID.randomUUID(),
                    device.getTenantId(),
                    device.getId(),
                    true,
                    expTime,
                    body
            );
            deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> {
                Optional<RpcError> rpcError = fromDeviceRpcResponse.getError();
                fromDeviceRpcResponse.getResponse();
                response.setData("ok");
                onDeviceRpcResponse( response);
                log.warn("[{}]",response);
            });


        }
    }

    private void onGatewayAttributesGet(MqttMessage message) {
        log.info("Configuration arrived! {}", message.toString());
        JsonNode payload = fromString(new String(message.getPayload(), StandardCharsets.UTF_8));
        if (payload.get("shared").get("configuration") != null) {
            String configuration = payload.get("shared").get("configuration").asText();
            if (!StringUtils.isEmpty(configuration)) {
                updateConfiguration(configuration);
            }
        }
    }

    private void onGatewayAttributesUpdate(MqttMessage message) {
        log.info("Configuration updates arrived! {}", message.toString());
        JsonNode payload = fromString(new String(message.getPayload(), StandardCharsets.UTF_8));
        if (payload.has("configuration")) {
            String configuration = payload.get("configuration").asText();
            if (!StringUtils.isEmpty(configuration)) {
                updateConfiguration(configuration);
            }
        }
    }

    private void updateConfiguration(String configuration) {
        try {
            onAppliedConfiguration(configuration);
        } catch (Exception e) {
            log.warn("Failed to update extension configurations");
        }
    }

    @Override
    public void onAppliedConfiguration(String configuration) {
        try {
            ObjectNode node = newNode();
            node.put("appliedConfiguration", configuration);
            MqttMessage msg = new MqttMessage(JsonTools.toString(node).getBytes(StandardCharsets.UTF_8));
            tbClient.publish(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, msg).waitForCompletion();
        } catch (Exception e) {
            log.warn("Could not publish applied configuration", e);
        }
    }


    private void onDeviceAttributesResponse(MqttMessage message) {
        JsonNode payload = fromString(new String(message.getPayload(), StandardCharsets.UTF_8));
        AttributeRequestKey requestKey = new AttributeRequestKey(payload.get("id").asInt(), payload.get("device").asText());
        AttributeRequestListener listener = pendingAttrRequestsMap.get(requestKey);
        if (listener == null) {
            log.warn("[{}][{}] Can't find listener for request", requestKey.getDeviceName(), requestKey.getRequestId());
            return;
        }

        AttributeRequest request = listener.getRequest();
        AttributeResponse.AttributeResponseBuilder response = AttributeResponse.builder();

        response.requestId(request.getRequestId());
        response.deviceName(request.getDeviceName());
        response.key(request.getAttributeKey());
        response.clientScope(request.isClientScope());
        response.topicExpression(request.getTopicExpression());
        response.valueExpression(request.getValueExpression());

        String key = listener.getRequest().getAttributeKey();
        JsonNode value = payload.get("value");
        if (value == null) {
            response.data(Optional.empty());
        } else if (value.isBoolean()) {
            response.data(Optional.of(new BooleanDataEntry(key, value.asBoolean())));
        } else if (value.isLong()) {
            response.data(Optional.of(new LongDataEntry(key, value.asLong())));
        } else if (value.isDouble()) {
            response.data(Optional.of(new DoubleDataEntry(key, value.asDouble())));
        } else {
            response.data(Optional.of(new StringDataEntry(key, value.asText())));
        }

        callbackExecutor.submit(() -> {
            try {
                listener.getListener().accept(response.build());
            } catch (Exception e) {
                log.error("[{}][{}] Failed to process attributes response", requestKey.getDeviceName(), requestKey.getRequestId(), e);
            }
        });
    }

    private void checkClientReconnected() {
        if (tbClient.isConnected()) {
            devices.forEach((k, v) -> onDeviceConnect(v.getName(), v.getType()));
        } else {
            scheduler.schedule(this::checkClientReconnected, CLIENT_RECONNECT_CHECK_INTERVAL, TimeUnit.SECONDS);
        }
    }
    private String fullName(String deviceName) {
        return configuration.getLabel()+"-"+deviceName;
    }

    private String realName(String deviceName) {
        return deviceName.replaceFirst(configuration.getLabel()+"-","");
    }
    private static String toString(Exception e) {
        StringWriter sw = new StringWriter();
        e.printStackTrace(new PrintWriter(sw));
        return sw.toString();
    }

    private <T> boolean subscribe(Function<T, Boolean> f, T sub) {
        if (f.apply(sub)) {
            log.info("Subscription added: {}", sub);
            return true;
        } else {
            log.warn("Subscription was already added: {}", sub);
            return false;
        }
    }

    private <T> boolean unsubscribe(Function<T, Boolean> f, T sub) {
        if (f.apply(sub)) {
            log.info("Subscription removed: {}", sub);
            return true;
        } else {
            log.warn("Subscription was already removed: {}", sub);
            return false;
        }
    }
}
